package data.email;

import data.Term;
import data.TermCollection;
import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInput;
import java.io.ObjectInputStream;
import java.io.ObjectOutput;
import java.io.ObjectOutputStream;
import java.io.OutputStream;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.Map;
import java.util.Queue;
import java.util.logging.Level;
import java.util.logging.Logger;
import process.DocumentStatisticsReader;

/**
 * EmailTermCollection adalah koleksi term dari dokumen email
 *
 * @author Robert Gunawan
 */
public class EmailTermCollection implements TermCollection {

    /**
     * integer iterator untuk runs file
     */
    private Integer runsIter = 0;
    /**
     * alamat dari term directory bagi worker, term directory berada di bawah
     * index directory
     */
    private String termDirectory;
    /**
     * lock untuk koleksi terms
     */
    private final Object terms_lock = new Object();
    /**
     * hashmap penampung term dokumen
     */
    private HashMap<String, Term> terms;
    /**
     * jumlah dokumen yang telah diekstrak
     */
    private Integer documentCount;

    /**
     * konstruktor dari email term collection
     */
    public EmailTermCollection() {
        terms = new HashMap<String, Term>();
    }

    @Override
    public void addTerm(Term term) {
        synchronized (terms_lock) {
            terms.put(term.getTermString(), term);
        }
    }

    @Override
    public boolean isExists(Term term) {
        synchronized (terms_lock) {
            return terms.containsKey(term.getTermString());
        }
    }

    @Override
    public synchronized void merge(Term term) {
        synchronized (terms_lock) {
            Term existingTerm = (Term) terms.get(term.getTermString());
            existingTerm.getStatistic().merge(term.getStatistic());
        }
    }

    @Override
    public Term getExistingTerm(String termString) {
        synchronized (terms_lock) {
            return (Term) terms.get(termString);
        }
    }

    @Override
    public HashMap<String, Term> getAllTerms() {
        synchronized (terms_lock) {
            return terms;
        }
    }

    @Override
    public Integer getDocumentCount() {
        return documentCount;
    }

    @Override
    public void setDocumentCount(Integer documentCount) {
        this.documentCount = documentCount;
    }

    @Override
    public void setTermDirectory(String directory) {
        this.termDirectory = directory;
    }

    @Override
    public String getTermDirectory() {
        return termDirectory;
    }

    @Override
    public void moveAllTerms(String newDirectory) {
        // baca semua file dalam directory
        File dir = new File(termDirectory);
        File subfolders[] = dir.listFiles();
        for (File subfolder : subfolders) {
            // baca file index dari tiap subfolder
            File indexFiles[] = subfolder.listFiles();
            for (File indexFile : indexFiles) {
                Term tempTerm = new Term(indexFile.getName());
                tempTerm.loadTerm(indexFile.getAbsolutePath());
                tempTerm.saveTerm(tempTerm.getPath(newDirectory));
            }
        }
    }

    @Override
    public void moveToFile(Term term) {
        synchronized (terms_lock) {
            if (terms.containsKey(term.getTermString())) {
                Term termRef = (Term) this.terms.get(term.getTermString());
                termRef.saveTerm(termRef.getPath(termDirectory));
                terms.remove(termRef.getTermString());
            }
        }
    }

    @Override
    public void moveAllToFile() {
        synchronized (terms_lock) {
            Iterator it = terms.entrySet().iterator();
            while (it.hasNext()) {
                Map.Entry pairs = (Map.Entry) it.next();
                Term term = (Term) pairs.getValue();
                term.saveTerm(term.getPath(termDirectory));
            }
        }
    }

    @Override
    public void moveAllToMemory() {
        File dirFile = new File(termDirectory);
        File directories[] = dirFile.listFiles();
        synchronized (terms_lock) {
            for (File directory : directories) {
                File indexFiles[] = directory.listFiles();
                for (File indexFile : indexFiles) {
                    Term newTerm = new Term(indexFile.getName());
                    newTerm.loadTerm(indexFile.getAbsolutePath());
                    terms.put(indexFile.getName(), newTerm);
                }
            }
        }
    }

    @Override
    public void readFromFile(Term term) {
        String filePath = term.getPath(termDirectory);
        File testFile = new File(filePath);
        if (testFile.exists()) {
            term.loadTerm(filePath);
            synchronized (terms_lock) {
                terms.put(term.getTermString(), term);
            }
        }
    }

    @Override
    public void purgeMemory() {
        synchronized (terms_lock) {
            terms.clear();
        }
    }

    @Override
    public void purgeDirectory() {
        File termDir = new File(termDirectory);
        File subfolders[] = termDir.listFiles();
        for (File subfolder : subfolders) {
            File indexFiles[] = subfolder.listFiles();
            for (File indexFile : indexFiles) {
                indexFile.delete();
            }
            subfolder.delete();
        }
        termDir.delete();
    }

    @Override
    public void writeToRuns() {
        runsIter++;
        String filePath = termDirectory + File.separator + "runs" + runsIter;
        System.out.println("## Writing terms to run file : " + filePath);
        try {
            OutputStream file = new FileOutputStream(filePath);
            OutputStream buffer = new BufferedOutputStream(file);
            ObjectOutput output = new ObjectOutputStream(buffer);
            try {
                output.writeObject(terms);
            } finally {
                output.close();
                buffer.close();
                file.close();
            }
        } catch (IOException ex) {
        }
    }

    @Override
    public void mergeAllRuns() {
        int i;
        System.out.println("## Merge all run file in " + termDirectory);
        File termDir = new File(termDirectory);
        File runFiles[] = termDir.listFiles();
        Queue<File> runFileQueue = new LinkedList<File>();
        for (File runFile : runFiles) {
            if (runFile.isFile()) {
                runFileQueue.add(runFile);
            }
        }
        while (!runFileQueue.isEmpty()) {
            this.purgeMemory();
            for (i = 0; (i < DocumentStatisticsReader.MAX_RUNS_IN_MEMORY) && (!runFileQueue.isEmpty()); i++) {
                File tempFile = runFileQueue.poll();
                mergeRunFile(tempFile.getAbsolutePath());
            }
            this.moveAllToFile();
        }

    }

    /**
     * memecah sebuah runs file untuk menjadi index file tiap term
     *
     * @param filePath file path ke runs file
     */
    public void mergeRunFile(String filePath) {
        try {
            System.out.println("## Merge run file : " + filePath);
            InputStream file = new FileInputStream(filePath);
            InputStream buffer = new BufferedInputStream(file);
            ObjectInput input = new ObjectInputStream(buffer);
            try {
                try {
                    HashMap<String, Term> tempTerms = (HashMap<String, Term>) input.readObject();
                    synchronized (terms_lock) {
                        Iterator it = tempTerms.entrySet().iterator();
                        while (it.hasNext()) {
                            Map.Entry pairs = (Map.Entry) it.next();
                            Term term = (Term) pairs.getValue();
                            if (this.isExists(term)) {
                                this.merge(term);
                            } else {
                                this.addTerm(term);
                            }
                        }
                    }
                } catch (ClassNotFoundException ex) {
                    Logger.getLogger(Term.class.getName()).log(Level.SEVERE, null, ex);
                }

            } finally {
                input.close();
                buffer.close();
                file.close();
                File runsFile = new File(filePath);
                runsFile.delete();
            }
        } catch (IOException ex) {
        }
    }
}
